package org.diysource

import java.util.Calendar

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.bean.SensorReading

import scala.util.Random

/**
 * description ：自定义数据源
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/21 19:22
 * modified By ：
 * version:    : 1.0
 */
class SensorSource extends RichParallelSourceFunction[SensorReading] {

  private var running = true

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]) = {
    // 创建一个随机器
    val random = new Random()

    // 创建10个传感器，初始化id,和温度
    val source = (1 to 10).map {
      i => (i, 65 + (random.nextGaussian() * 20))
    }

    // 动态修改传感器的温度并将记录写出
    while (running) {
      source.foreach(
        m => {
          sourceContext.collect(SensorReading("" + m._1, Calendar.getInstance().getTimeInMillis, m._2 + (random.nextGaussian() * 0.5))
          )
        }
      )
     Thread.sleep(100)
    }
  }

  override def cancel() = running = false
}
